package com.zhaotianhao.check.comboserver.kafka;

import cn.hutool.http.HttpRequest;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.zhaotianhao.check.comboserver.entity.Combo;
import com.zhaotianhao.check.comboserver.entity.GroupCombo;
import com.zhaotianhao.check.comboserver.mapper.ComboRespository;
import com.zhaotianhao.check.comboserver.service.IComboService;
import com.zhaotianhao.check.comboserver.service.IGroupComboService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author Z-7
 * @Date 2022/4/3
 */
@Slf4j
@Component
public class Kafka implements MessageListener<String, String> {
    @Resource
    private IComboService iComboService;
    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private IGroupComboService iGroupComboService;
    @Resource
    private ComboRespository comboRespository;

    @Override
    @KafkaListener(topics = "comboAdd")
    public void onMessage(ConsumerRecord<String, String> data) {
        String key = data.key();
        if (!redisTemplate.opsForValue().setIfAbsent(key,"",5, TimeUnit.HOURS)) {
            log.info("消息重复消费");
            return;
        }
        log.info("------------------------------------------------------------------");
        log.info("kafka正常消费");
        Combo combo = iComboService.getById(data.value());
        QueryWrapper<GroupCombo> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("combo_id",combo.getId());
        List<GroupCombo> groupCombos = iGroupComboService.list(queryWrapper);
        List<Integer> goodIds = groupCombos.stream().map(groupCombo -> groupCombo.getGroupId()).collect(Collectors.toList());
        Map<String, Object> map = new HashMap<>();
        map.put("ids",goodIds);
        String groups = HttpRequest.post("http://192.168.127.142:8083/group/comboGroups")
                .body(JSONObject.toJSONString(map))
                .execute()
                .body();
        String[] split = groups.split("\\|");
        combo.setGroupNames(split[0]);
        combo.setItemNames(split[1]);
        // 存入es
        comboRespository.save(combo);
        // 删除reids缓存
        redisTemplate.delete("message:"+key);
    }
}
